// Copyright 2023 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#ifndef HEARTBEAT_H
#define HEARTBEAT_H

#include <deque>
#include <thread>
#include <functional>
#include "Communication/InterThreadComm.h"
#include "Communication/MsgFragment/MsgFragment.h"
#include "Communication/MsgFragment/MsgReassemble.h"
#include "Communication/Socket/VmiSocket.h"
#include "Communication/StreamParse/StreamParser.h"
#include "Include/msghead.h"

namespace {
    const int BYTE_ALIGNMENT = 7;
}

struct HeartbeatMsg {
    uint64_t sendMicroSecond;
    uint64_t seqNum;
    uint8_t type;
    uint8_t pad[BYTE_ALIGNMENT];
} __attribute__((packed));

class Heartbeat : public ServiceHandle {
public:
    class ThreadPara {
    public:
        /**
         * @brief: thread need param in this, quarantine from start thread
         */
        ThreadPara();

        /**
         * @brief: send heartbeat request
         * @param [in] StreamMsgHead*: heartbeat message
         */
        void SendHeartbeatRequest(StreamMsgHead& head);

        /**
         * @brief: handle recv heartbeat
         */
        void RecvHeartbeatResponse();

        /**
         * @brief: check net status
         */
        void CheckNetStatus();

        /**
         * @brief: handle recv heartbeat
         * @param [in] reassembleMsg: heartbeat message
         * @param [in] recvTime: recvTime
         */
        void HandleRecvMsg(const std::pair<uint8_t*, uint32_t>& reassembleMsg, struct timespec recvTime);

        /**
         * @brief: set m_nextSendTime
         * @param [in] nextTime: nextTime
         */
        void SetNextSendTime(uint64_t nextTime);

        /**
         * @brief: get m_netLoopbackLatencyAverage
         * @return: m_netLoopbackLatencyAverage
         */
        uint64_t GetNetLoopbackLatencyAverage();

        /**
         * @brief: get net loop max
         * @return: loop, 0 is Infinity
         */
        uint64_t GetNetLoopbackLatencyMax();

        /**
         * @brief: get m_packetQueue
         * @return: m_packetQueue
         */
        InterThreadComm& GetInterThreadComm();

        /**
         * @brief: get m_stopFlag
         * @return: m_stopFlag
         */
        bool GetStopFlag();

        /**
         * @brief: set m_stopFlag
         * @param [in] stopFlag: stopFlag
         */
        void SetStopFlag(bool stopFlag);

        /**
         * @brief: set m_vmiSocket
         * @param [in] vmiSocket: vmiSocket
         */
        void SetVmiSocket(std::shared_ptr<VmiSocket> vmiSocket);

        /**
         * @brief: get m_vmiSocket
         * @return: m_vmiSocket
         */
        std::shared_ptr<VmiSocket> GetVmiSocket();

        /**
         * @brief: set m_callback
         * @param [in] callback: callback
         */
        void SetCallback(std::function<void(std::shared_ptr<VmiSocket>)> callback);

    private:

        /**
         * @brief: compute net loop
         * @param [in] pHeartbeatResMsg: heartbeat package
         * @param [in] recvTime: recvTime
         */
        void CalcLatency(HeartbeatMsg& pHeartbeatResMsg, uint64_t recvTime);
        uint64_t m_nextSendTime;     // mircosecond
        uint64_t m_netLoopbackLatencyAverage;
        uint64_t m_netLoopbackLatencyMax;
        std::deque<std::pair<uint64_t, uint64_t>> m_latencyQueue = {};
        InterThreadComm m_packetQueue{false, false, true};
        volatile bool m_stopFlag;
        std::shared_ptr<VmiSocket> m_vmiSocket = nullptr;
        uint64_t m_sendSeqNum;
        uint64_t m_recvSeqNum;
        MsgFragment m_msgFragment = {};
        MsgReassemble m_msgReassemble{VMIMsgType::HEARTBEAT};
        std::function<void(std::shared_ptr<VmiSocket>)> m_callback = nullptr;
    };

    /**
     * @brirf: construct
     * @param [in] sock: socket
     * @param [in] fnCallback: check net exception, callbak it
     */
    Heartbeat(std::shared_ptr<VmiSocket> sock, std::function<void(std::shared_ptr<VmiSocket>)> fnCallback);

    ~Heartbeat() override;

    /**
     * @brief: start thread
     */
    bool Start();

    /**
     * @brief: stop thread, caller wait thread exit in hear.
     */
    void ForceStop();

    /**
     * @brief: put heartbeat data into queue
     * @param [in] packet: heartbeat data
     */
    void Handle(const VmiBuffer& packet) override;

    /**
     * @brief: Get network average latency
     * @return: latency, 0 is Infinity
     */
    uint64_t GetNetLoopbackLatencyAverage();

    /**
     * @brief: Get network maximum latency
     * @return: latency, 0 is Infinity
     */
    uint64_t GetNetLoopbackLatencyMax();

private:
    std::mutex m_lock = {};
    std::shared_ptr<VmiSocket> m_vmiSocket = nullptr;
    std::function<void(std::shared_ptr<VmiSocket>)> m_callback = nullptr;
    std::thread *m_task = nullptr;
    std::shared_ptr<ThreadPara> m_threadParameter = nullptr;

    /**
     * @brief: heartbeat has for function:
     *               1、timing send heartbeat (SendHeartbeatRequest);
     *               2、handle recv heartbeat (RecvHeartbeatResponse);
     *               3、compute net loop;
     *               4、check net status.
     * @param [in] param: heartbeat param
     */
    friend void TaskentryHeartbeat(std::shared_ptr<Heartbeat::ThreadPara> param);
};

#endif
